package org.wj.config.temperature;

import cn.hutool.core.io.FileUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.net.URI;
import java.nio.file.Paths;

public class SecondarySortDriver extends Configured implements Tool {

    public static void main(String[] args) throws Exception {

        try {
            FileUtil.del(Paths.get("/tmp"));
            int run = ToolRunner.run(new SecondarySortDriver(), args);
            System.exit(run);
        } finally {
            FileUtil.del(Paths.get("/tmp"));
        }

    }

    @Override
    public int run(String[] strings) throws Exception {
        Configuration conf = getConf();
//        conf.set("mapreduce.framework.name", "local");
        JobConf job = new JobConf(conf);
        job.setJarByClass(SecondarySortDriver.class);
        job.setJobName("temperature01");

        Path inputPath = new Path("hdfs://localhost:9000/temperature.txt");
        Path outputPath = new Path("hdfs://localhost:9000/temperatureOut");
        FileInputFormat.setInputPaths(job, inputPath);
        FileSystem fileSystem = FileSystem.newInstance(new URI("hdfs://localhost:9000"), conf);
        if (fileSystem.exists(outputPath)) {
            fileSystem.delete(outputPath, true);
        }

        FileOutputFormat.setOutputPath(job, outputPath);

        job.setMapOutputKeyClass(DateTemperaturePair.class);
        job.setMapOutputValueClass(LongWritable.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        job.setMapperClass(SecondarySortMapper.class);
        job.setReducerClass(SecondarySortReducer.class);
        job.setPartitionerClass(DateTemperaturePartitioner.class);
        job.setOutputValueGroupingComparator(DateTemperatureGroupingComparator.class);
        RunningJob runningJob = JobClient.runJob(job);
        runningJob.waitForCompletion();
        boolean complete = runningJob.isComplete();
        return complete ? 0 : 1;
    }
}
